3 use MediaWiki\MediaWikiServices
;
10 class JobQueueTest
extends MediaWikiTestCase
{
12 protected $queueRand, $queueRandTTL, $queueFifo, $queueFifoTTL;
14 function __construct( $name = null, array $data = [], $dataName = '' ) {
15 parent
::__construct( $name, $data, $dataName );
17 $this->tablesUsed
[] = 'job';
20 protected function setUp() {
21 global $wgJobTypeConf;
24 if ( $this->getCliArg( 'use-jobqueue' ) ) {
25 $name = $this->getCliArg( 'use-jobqueue' );
26 if ( !isset( $wgJobTypeConf[$name] ) ) {
27 throw new MWException( "No \$wgJobTypeConf entry for '$name'." );
29 $baseConfig = $wgJobTypeConf[$name];
31 $baseConfig = [ 'class' => JobQueueDBSingle
::class ];
33 $baseConfig['type'] = 'null';
34 $baseConfig['wiki'] = wfWikiID();
36 'queueRand' => [ 'order' => 'random', 'claimTTL' => 0 ],
37 'queueRandTTL' => [ 'order' => 'random', 'claimTTL' => 10 ],
38 'queueTimestamp' => [ 'order' => 'timestamp', 'claimTTL' => 0 ],
39 'queueTimestampTTL' => [ 'order' => 'timestamp', 'claimTTL' => 10 ],
40 'queueFifo' => [ 'order' => 'fifo', 'claimTTL' => 0 ],
41 'queueFifoTTL' => [ 'order' => 'fifo', 'claimTTL' => 10 ],
43 foreach ( $variants as $q => $settings ) {
45 $this->$q = JobQueue
::factory( $settings +
$baseConfig );
46 } catch ( MWException
$e ) {
48 // @todo What if it was another error?
53 protected function tearDown() {
57 'queueRand', 'queueRandTTL', 'queueTimestamp', 'queueTimestampTTL',
58 'queueFifo', 'queueFifoTTL'
69 * @dataProvider provider_queueLists
70 * @covers JobQueue::getWiki
72 public function testGetWiki( $queue, $recycles, $desc ) {
73 $queue = $this->$queue;
75 $this->markTestSkipped( $desc );
77 $this->assertEquals( wfWikiID(), $queue->getWiki(), "Proper wiki ID ($desc)" );
78 $this->assertEquals( wfWikiID(), $queue->getDomain(), "Proper wiki ID ($desc)" );
82 * @dataProvider provider_queueLists
83 * @covers JobQueue::getType
85 public function testGetType( $queue, $recycles, $desc ) {
86 $queue = $this->$queue;
88 $this->markTestSkipped( $desc );
90 $this->assertEquals( 'null', $queue->getType(), "Proper job type ($desc)" );
94 * @dataProvider provider_queueLists
97 public function testBasicOperations( $queue, $recycles, $desc ) {
98 $queue = $this->$queue;
100 $this->markTestSkipped( $desc );
103 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
105 $queue->flushCaches();
106 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
107 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
109 $this->assertNull( $queue->push( $this->newJob() ), "Push worked ($desc)" );
110 $this->assertNull( $queue->batchPush( [ $this->newJob() ] ), "Push worked ($desc)" );
112 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
114 $queue->flushCaches();
115 $this->assertEquals( 2, $queue->getSize(), "Queue size is correct ($desc)" );
116 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
117 $jobs = iterator_to_array( $queue->getAllQueuedJobs() );
118 $this->assertEquals( 2, count( $jobs ), "Queue iterator size is correct ($desc)" );
120 $job1 = $queue->pop();
121 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
123 $queue->flushCaches();
124 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
126 $queue->flushCaches();
128 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
131 $job2 = $queue->pop();
132 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
133 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
135 $queue->flushCaches();
137 $this->assertEquals( 2, $queue->getAcquiredCount(), "Active job count ($desc)" );
140 $queue->ack( $job1 );
142 $queue->flushCaches();
144 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
147 $queue->ack( $job2 );
149 $queue->flushCaches();
150 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
152 $this->assertNull( $queue->batchPush( [ $this->newJob(), $this->newJob() ] ),
153 "Push worked ($desc)" );
154 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
157 $queue->flushCaches();
158 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
159 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
163 * @dataProvider provider_queueLists
166 public function testBasicDeduplication( $queue, $recycles, $desc ) {
167 $queue = $this->$queue;
169 $this->markTestSkipped( $desc );
172 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
174 $queue->flushCaches();
175 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
176 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
180 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
182 "Push worked ($desc)" );
184 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
186 $queue->flushCaches();
187 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
188 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
192 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
194 "Push worked ($desc)"
197 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
199 $queue->flushCaches();
200 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
201 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
203 $job1 = $queue->pop();
204 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
206 $queue->flushCaches();
207 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
209 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
212 $queue->ack( $job1 );
214 $queue->flushCaches();
215 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
219 * @dataProvider provider_queueLists
222 public function testDeduplicationWhileClaimed( $queue, $recycles, $desc ) {
223 $queue = $this->$queue;
225 $this->markTestSkipped( $desc );
228 $job = $this->newDedupedJob();
229 $queue->push( $job );
231 // De-duplication does not apply to already-claimed jobs
233 $queue->push( $job );
237 // Make sure ack() of the twin did not delete the sibling data
238 $this->assertType( NullJob
::class, $j );
242 * @dataProvider provider_queueLists
245 public function testRootDeduplication( $queue, $recycles, $desc ) {
246 $queue = $this->$queue;
248 $this->markTestSkipped( $desc );
251 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
253 $queue->flushCaches();
254 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
255 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
257 $id = wfRandomString( 32 );
258 $root1 = Job
::newRootJobParams( "nulljobspam:$id" ); // task ID/timestamp
259 for ( $i = 0; $i < 5; ++
$i ) {
260 $this->assertNull( $queue->push( $this->newJob( 0, $root1 ) ), "Push worked ($desc)" );
262 $queue->deduplicateRootJob( $this->newJob( 0, $root1 ) );
265 # Add a second to UNIX epoch and format back to TS_MW
266 $root2_ts = strtotime( $root2['rootJobTimestamp'] );
268 $root2['rootJobTimestamp'] = wfTimestamp( TS_MW
, $root2_ts );
270 $this->assertNotEquals( $root1['rootJobTimestamp'], $root2['rootJobTimestamp'],
271 "Root job signatures have different timestamps." );
272 for ( $i = 0; $i < 5; ++
$i ) {
273 $this->assertNull( $queue->push( $this->newJob( 0, $root2 ) ), "Push worked ($desc)" );
275 $queue->deduplicateRootJob( $this->newJob( 0, $root2 ) );
277 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
279 $queue->flushCaches();
280 $this->assertEquals( 10, $queue->getSize(), "Queue size is correct ($desc)" );
281 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
286 $job = $queue->pop();
291 if ( $job instanceof DuplicateJob
) {
296 $this->assertEquals( 10, count( $jobs ), "Correct number of jobs popped ($desc)" );
297 $this->assertEquals( 5, $dupcount, "Correct number of duplicate jobs popped ($desc)" );
301 * @dataProvider provider_fifoQueueLists
304 public function testJobOrder( $queue, $recycles, $desc ) {
305 $queue = $this->$queue;
307 $this->markTestSkipped( $desc );
310 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
312 $queue->flushCaches();
313 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
314 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
316 for ( $i = 0; $i < 10; ++
$i ) {
317 $this->assertNull( $queue->push( $this->newJob( $i ) ), "Push worked ($desc)" );
320 for ( $i = 0; $i < 10; ++
$i ) {
321 $job = $queue->pop();
322 $this->assertTrue( $job instanceof Job
, "Jobs popped from queue ($desc)" );
323 $params = $job->getParams();
324 $this->assertEquals( $i, $params['i'], "Job popped from queue is FIFO ($desc)" );
328 $this->assertFalse( $queue->pop(), "Queue is not empty ($desc)" );
330 $queue->flushCaches();
331 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
332 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
338 public function testQueueAggregateTable() {
339 $queue = $this->queueFifo
;
340 if ( !$queue ||
!method_exists( $queue, 'getServerQueuesWithJobs' ) ) {
341 $this->markTestSkipped();
344 $this->assertNotContains(
345 [ $queue->getType(), $queue->getWiki() ],
346 $queue->getServerQueuesWithJobs(),
347 "Null queue not in listing"
350 $queue->push( $this->newJob( 0 ) );
352 $this->assertContains(
353 [ $queue->getType(), $queue->getWiki() ],
354 $queue->getServerQueuesWithJobs(),
355 "Null queue in listing"
359 public static function provider_queueLists() {
361 [ 'queueRand', false, 'Random queue without ack()' ],
362 [ 'queueRandTTL', true, 'Random queue with ack()' ],
363 [ 'queueTimestamp', false, 'Time ordered queue without ack()' ],
364 [ 'queueTimestampTTL', true, 'Time ordered queue with ack()' ],
365 [ 'queueFifo', false, 'FIFO ordered queue without ack()' ],
366 [ 'queueFifoTTL', true, 'FIFO ordered queue with ack()' ]
370 public static function provider_fifoQueueLists() {
372 [ 'queueFifo', false, 'Ordered queue without ack()' ],
373 [ 'queueFifoTTL', true, 'Ordered queue with ack()' ]
377 function newJob( $i = 0, $rootJob = [] ) {
378 return new NullJob( Title
::newMainPage(),
379 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 0, 'i' => $i ] +
$rootJob );
382 function newDedupedJob( $i = 0, $rootJob = [] ) {
383 return new NullJob( Title
::newMainPage(),
384 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 1, 'i' => $i ] +
$rootJob );
388 class JobQueueDBSingle
extends JobQueueDB
{
389 protected function getDB( $index ) {
390 $lb = MediaWikiServices
::getInstance()->getDBLoadBalancer();
391 // Override to not use CONN_TRX_AUTOCOMMIT so that we see the same temporary `job` table
392 return $lb->getConnection( $index, [], $this->domain
);